package com.flink.streaming.web.util;

import com.alibaba.fastjson.JSONObject;
import com.flink.streaming.web.config.AzKaBanConfig;
import com.flink.streaming.web.exceptions.BizException;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.FileSystemResource;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.stereotype.Component;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StringUtils;

import java.io.File;
import java.util.Map;
import java.util.Objects;

@Component
public class AzKaBanUtil {

    /**
     * azkaban成功状态
     **/
    public final static String AZK_SUCCESS = "success";

    private static final String X_REQUESTED_WITH = "XMLHttpRequest";

    private static final Logger logger = LoggerFactory.getLogger(AzKaBanUtil.class);

    private static final String CONTENT_TYPE = "application/x-www-form-urlencoded; charset=utf-8";


    /**
     * Azkaban登录接口,返回sessionId
     * @return
     * @throws Exception
     */
    public static String login(AzKaBanConfig azKaBanConfig) throws Exception {

        HttpHeaders hs = new HttpHeaders();
        hs.add("Content-Type", CONTENT_TYPE);
        hs.add("X-Requested-With", X_REQUESTED_WITH);
        LinkedMultiValueMap<String, String> linkedMultiValueMap = new LinkedMultiValueMap<String, String>();
        linkedMultiValueMap.add("action", "login");
        linkedMultiValueMap.add("username", azKaBanConfig.getAzkUsername());
        linkedMultiValueMap.add("password", azKaBanConfig.getAzkPassword());
        HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(linkedMultiValueMap, hs);

        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl(), httpEntity, String.class);

        JSONObject resultJsonObject = JSONObject.parseObject(result);

        if (!AZK_SUCCESS.equals(resultJsonObject.get("status"))) {
            logger.error("Azkaban登录失败!返回错误信息：" + result);
            throw new BizException("Azkaban登录失败!");
        }

        return resultJsonObject.get("session.id").toString();
    }


    /**
     * Azkaban上传zip文件
     * @param projectName 项目名称
     * @param file        文件
     * @return projectId编号
     * @throws Exception
     */
    public static String uploadZip(String projectName, File file, AzKaBanConfig azKaBanConfig, String sessionId) throws Exception {
        return doUploadZIP(projectName, file, azKaBanConfig, sessionId);
    }

    private static String doUploadZIP(String projectName, File file, AzKaBanConfig azKaBanConfig, String sessionId) {
        FileSystemResource resource = new FileSystemResource(file);
        LinkedMultiValueMap<String, Object> linkedMultiValueMap = new LinkedMultiValueMap<String, Object>();
        linkedMultiValueMap.add("ajax", "upload");
        linkedMultiValueMap.add("file", resource);
        linkedMultiValueMap.add("project", projectName);
        linkedMultiValueMap.add("session.id", sessionId);
        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl() + "/manager", linkedMultiValueMap, String.class);

        JSONObject resultJsonObject = JSONObject.parseObject(result);

        if (StringUtils.isEmpty(resultJsonObject.get("projectId"))) {
            logger.error("上传文件至Azkaban失败:", projectName, file.getPath());
            logger.error("Azkaban上传文件失败!返回错误信息：" + result);
            throw new BizException("Azkaban上传文件失败!返回错误信息：" + result);
        }
        return resultJsonObject.getString("projectId");
    }

    public static String scheduleByCronEXEaFlow(String projectName, String sessionId, String cron, String flowName, AzKaBanConfig azKaBanConfig) {

        HttpHeaders hs = new HttpHeaders();
        hs.add("Content-Type", CONTENT_TYPE);
        hs.add("X-Requested-With", X_REQUESTED_WITH);
        LinkedMultiValueMap<String, String> linkedMultiValueMap = new LinkedMultiValueMap<>();
        linkedMultiValueMap.add("flow", flowName);
        linkedMultiValueMap.add("cronExpression", cron);
        linkedMultiValueMap.add("session.id", sessionId);
        linkedMultiValueMap.add("projectName", projectName);
        linkedMultiValueMap.add("ajax", "scheduleCronFlow");

        HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(linkedMultiValueMap, hs);
        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl() + "/schedule", httpEntity, String.class);

        JSONObject resultJsonObject = JSONObject.parseObject(result);

        if (StringUtils.isEmpty(resultJsonObject.get("scheduleId"))) {
            logger.error("调度失败信息： projectName = {}， flowName = {}， cron = {}", projectName, flowName, cron);
            logger.error("Azkaban返回根据时间创建定时任务信息失败：!返回错误信息" + result);
            throw new BizException("Azkaban返回根据时间创建定时任务信息失败：!返回错误信息" + result);
        }
        return resultJsonObject.getString("scheduleId");
    }

    public static void unscheduleAFlow(String scheduleId, String sessionID, AzKaBanConfig azKaBanConfig) {
        HttpHeaders hs = new HttpHeaders();
        hs.add("Content-Type", CONTENT_TYPE);
        hs.add("X-Requested-With", X_REQUESTED_WITH);
        LinkedMultiValueMap<String, String> linkedMultiValueMap = new LinkedMultiValueMap<>();
        linkedMultiValueMap.add("session.id", sessionID);
        linkedMultiValueMap.add("scheduleId", scheduleId);
        linkedMultiValueMap.add("action", "removeSched");

        HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(linkedMultiValueMap, hs);
        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl() + "/schedule", httpEntity, String.class);

        JSONObject resultJsonObject = JSONObject.parseObject(result);

        if (!AZK_SUCCESS.equals(resultJsonObject.get("status"))) {
            logger.error("取消调度失败 调度id={}", scheduleId);
            logger.error("Azkaban返回根据时间创建定时任务信息失败：!返回错误信息" + result);
        }
    }

    public static String fetchASchedule(String projectId, String sessionID, String flowId, AzKaBanConfig azKaBanConfig) {

        HttpHeaders hs = new HttpHeaders();
        hs.add("Content-Type", CONTENT_TYPE);
        hs.add("X-Requested-With", X_REQUESTED_WITH);
        LinkedMultiValueMap<String, String> linkedMultiValueMap = new LinkedMultiValueMap<>();
        linkedMultiValueMap.add("flowId", flowId);
        linkedMultiValueMap.add("session.id", sessionID);
        linkedMultiValueMap.add("projectId", projectId);
        linkedMultiValueMap.add("ajax", "fetchSchedule");

        HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(linkedMultiValueMap, hs);
        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl() + "/schedule", httpEntity, String.class);

        JSONObject resultJsonObject = JSONObject.parseObject(result);

        if (Objects.isNull(resultJsonObject.get("schedule"))) {
            logger.error("获取调度计划表失败 项目id={}, flowName={}", projectId, flowId);
            logger.error("Azkaban返回根据时间创建定时任务信息失败：!返回错误信息" + result);
            return null;
        }
        JSONObject schedule = resultJsonObject.getJSONObject("schedule");
        return schedule.getString("scheduleId");
    }

    public static String fetchProjectId(String projectName, String sessionID, AzKaBanConfig azKaBanConfig) {

        HttpHeaders hs = new HttpHeaders();
        hs.add("Content-Type", CONTENT_TYPE);
        hs.add("X-Requested-With", X_REQUESTED_WITH);
        LinkedMultiValueMap<String, String> linkedMultiValueMap = new LinkedMultiValueMap<>();
        linkedMultiValueMap.add("project", projectName);
        linkedMultiValueMap.add("session.id", sessionID);
        linkedMultiValueMap.add("ajax", "fetchprojectflows");

        HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(linkedMultiValueMap, hs);
        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl() + "/schedule", httpEntity, String.class);

        JSONObject resultJsonObject = JSONObject.parseObject(result);

        if (Objects.isNull(resultJsonObject.get("projectId"))) {
            logger.error("获取调度计划表失败 项目名称={}", projectName);
            logger.error("Azkaban获取项目信息失败!返回错误信息" + result);
            throw new BizException("Azkaban获取项目信息失败：!返回错误信息" + result);
        }
        return resultJsonObject.getString("projectId");
    }


    public static String executeFlow(String projectName, String flowName, String sessionID, AzKaBanConfig azKaBanConfig, Map<String, String> args) {
        HttpHeaders hs = new HttpHeaders();
        hs.add("Content-Type", CONTENT_TYPE);
        hs.add("X-Requested-With", X_REQUESTED_WITH);
        LinkedMultiValueMap<String, String> linkedMultiValueMap = new LinkedMultiValueMap<>();
        linkedMultiValueMap.add("flow", flowName);
        linkedMultiValueMap.add("ajax", "executeFlow");
        linkedMultiValueMap.add("project", projectName);
        linkedMultiValueMap.add("session.id", sessionID);
        args.forEach((s, s2) -> linkedMultiValueMap.add("flowOverride[" + s + "]", s2));
        HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(linkedMultiValueMap, hs);
        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl() + "/executor", httpEntity, String.class);

        JSONObject resultJsonObject = JSONObject.parseObject(result);

        // 返回执行id
        if (Objects.isNull(resultJsonObject.get("execid"))) {
            logger.error("执行调度计划表失败 项目名称={}, flow={}, 参数={}", projectName, flowName, JSONObject.toJSON(args));
            logger.error("Azkaban执行调度计划表失败 !返回错误信息" + result);
            throw new BizException("Azkaban执行调度计划表失败 ：!返回错误信息" + result);
        }
        return resultJsonObject.getString("execid");
    }


    public static JSONObject fetchExecJobLogs(Long offset, Integer length, String execId, String sessionId, AzKaBanConfig azKaBanConfig) {
        HttpHeaders hs = new HttpHeaders();
        hs.add("Content-Type", CONTENT_TYPE);
        hs.add("X-Requested-With", X_REQUESTED_WITH);
        LinkedMultiValueMap<String, Object> linkedMultiValueMap = new LinkedMultiValueMap<>();
        linkedMultiValueMap.add("execid", execId);
        linkedMultiValueMap.add("offset", offset);
        linkedMultiValueMap.add("length", length);
        linkedMultiValueMap.add("ajax", "fetchExecFlowLogs");
        linkedMultiValueMap.add("session.id", sessionId);
        HttpEntity<MultiValueMap<String, Object>> httpEntity = new HttpEntity<>(linkedMultiValueMap, hs);
        String result = azKaBanConfig.getRestTemplate().postForObject(azKaBanConfig.getAzkUrl() + "/executor", httpEntity, String.class);
        return JSONObject.parseObject(result);
    }
}
